package org.im.threadpool.disruptor;

import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

import com.concurrent.execute.threadPool.DaemonThreadFactory;
import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

public class TaskProducer {
	private final RingBuffer<Task> ringBuffer;

	public TaskProducer(RingBuffer<Task> ringBuffer) {
		this.ringBuffer = ringBuffer;
	}

	public void onData(Object data) {
		long sequence = ringBuffer.next(); // Grab the next sequence
		try {
			Task event = ringBuffer.get(sequence); // Get the entry in the Disruptor
													// for the sequence
			event.setData(data);
		} finally {
			ringBuffer.publish(sequence);
		}
	}

	public static void main(String[] args) throws Exception {
		// Executor that will be used to construct new threads for consumers
		Executor executor = Executors.newCachedThreadPool();

		// The factory for the event
		TaskFacotry factory = new TaskFacotry();

		// Specify the size of the ring buffer, must be power of 2.
		int bufferSize = 1024;

		// Construct the Disruptor
		Disruptor<Task> disruptor = new Disruptor<>(factory, bufferSize, new DaemonThreadFactory(), ProducerType.MULTI,
				new BlockingWaitStrategy());

		// Connect the handler
		disruptor.handleEventsWith(new TaskExector());

		// Start the Disruptor, starts all threads running
		disruptor.start();

		// Get the ring buffer from the Disruptor to be used for publishing.
		RingBuffer<Task> ringBuffer = disruptor.getRingBuffer();

		final TaskProducer producer = new TaskProducer(ringBuffer);

		for (int i = 0; i < 5; i++) {
			new Thread() {
				public void run() {
					String helloWorld = "hello world";
					for (long l = 0; l < 1000000; l++) {
						producer.onData(helloWorld + l);
						// Thread.sleep(1000);
					}
				}
			}.start();
		}

	}
}
